package com.spark.kafka

import java.sql
import java.sql.DriverManager

import org.apache.spark.SparkConf
import org.apache.spark.streaming._
import org.apache.spark.streaming.kafka.KafkaUtils

/**
  * Created by hjn on 2020.3.8
  */

object insertIntoMySQL {
  def main(args: Array[String]): Unit = {
    // 设置console提示信息等级，过滤INFO信息
    //    StreamingExamples.setStreamingLogLevels()

    /**
      * Spark的运行模式取决于传递给SparkContext的Master URL的值。Master URL可以是以下任一种形式：
      * local 使用一个Worker线程本地化运行SPARK(完全不并行)
      * local[*] 使用逻辑CPU个数数量的线程来本地化运行Spark
      * local[K] 使用K个Worker线程本地化运行Spark（理想情况下，K应该根据运行机器的CPU核数设定）
      * spark://HOST:PORT 连接到指定的Spark standalone master。默认端口是7077.
      * yarn-client 以客户端模式连接YARN集群。集群的位置可以在HADOOP_CONF_DIR 环境变量中找到。
      * yarn-cluster 以集群模式连接YARN集群。集群的位置可以在HADOOP_CONF_DIR 环境变量中找到。
      * mesos://HOST:PORT 连接到指定的Mesos集群。默认接口是5050。
      */
    //val sc = new SparkConf().setAppName("insertIntoMySQL").setMaster("yarn-client")
    val sc = new SparkConf().setAppName("insertIntoMySQL").setMaster("local[*]")
    val ssc = new StreamingContext(sc, Milliseconds(100))
    ssc.sparkContext.setLogLevel("ERROR")
    //设置检查点，如果存放在HDFS上面，则写成类似ssc.checkpoint("/user/hadoop/checkpoint")这种形式，但是，要启动hadoop
    //ssc.checkpoint("file:///usr/local/spark/mycode/kafka/checkpoint")

    // zookeeper kafka 配置信息
    val zkQuorum = "localhost:2181"
    val group = "1" //topic所在的group，可以设置为自己想要的名称，比如不用1，而是val group = "test-consumer-group"
    val topics = "mobileData"
    val numThreads = 1 // 每个topic的分区数
    val topicMap = topics.split(",").map((_, numThreads.toInt)).toMap

    // 接收kafka数据
    val lineMap = KafkaUtils.createStream(ssc, zkQuorum, group, topicMap)

    val lines = lineMap.map(_._2)

    lines.foreachRDD(line => {
      //内部函数
      def func(records: Iterator[String]) {
        var conn: sql.Connection = null
        var stmt: sql.PreparedStatement = null
        try {
          val url = "jdbc:mysql://localhost:3306/test?useUnicode=true&characterEncoding=utf-8&useSSL=false"
          val user = "root"
          val password = "root" //数据库密码是root，请改成你自己的mysql数据库密码
          conn = DriverManager.getConnection(url, user, password)
          records.foreach(p => {
            val arr = p.split(",")
            val date = arr(0)
            val imsi = arr(1)
            val laci = arr(2)
            val longtitude = arr(3)
            val latitude = arr(4)
            var sql = "insert into mobileData(date, imsi, laci, longitude, latitude) values (?,?,?,?,?)"
            stmt = conn.prepareStatement(sql);
            stmt.setString(1, date)
            stmt.setString(2, imsi)
            stmt.setString(3, laci)
            stmt.setDouble(4, longtitude.toDouble)
            stmt.setDouble(5, latitude.toDouble)
            stmt.executeUpdate()

            printf("insert into mobileData(date, imsi, laci, longitude, latitude) values (%s,%s,%s,%s,%s)", date, imsi, laci, longtitude, latitude)
            println()
            println()
          })

        } catch {
          case e: Exception => e.printStackTrace()
            conn.rollback()
        } finally {
          if (stmt != null) {
            stmt.close()
          }
          if (conn != null) {
            conn.close()
          }
        }
      }

      val repartitionedRDD = line.repartition(1)
      repartitionedRDD.foreachPartition(func)
    })

    ssc.start()
    ssc.awaitTermination()
  }
}